package KafkaTwitterProducer;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.Properties;

import twitter4j.*;
import twitter4j.conf.ConfigurationBuilder;

import java.util.List;

/**
 * Created by Ljc on 5/4/2017.
 */


public class KafkaTwitterProducer {
	
	public static Producer<String,String> init(){
		
        // create instance for properties to access producer configs
        Properties props = new Properties();

        //Assign localhost id
        //props.put("bootstrap.servers", "student27-x1:9092,student27-x2:9092,student28-x1:9092,student28-x2:9092");
        props.put("bootstrap.servers", "student39-x1:9092,student39-x2:9092,student40-x1:9092,student40-x2:9092");
        
        //Set acknowledgements for producer requests.
        props.put("acks", "all");

        //If the request fails, the producer can automatically retry,
        props.put("retries", 0);

        //Specify buffer size in config
        props.put("batch.size", 16384);

        //Reduce the no of requests less than 0
        props.put("linger.ms", 1);

        //The buffer.memory controls the total amount of memory available to the producer for buffering.
        props.put("buffer.memory", 33554432);

        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        return producer;
	}
	
    public static void main(String[] args) throws Exception{


    	String topicName = "sparktwitter";
        
        
        Producer<String, String> producer = init();
        
        
        //set up key to Twitter API 
        ConfigurationBuilder cb = new ConfigurationBuilder();
        cb.setDebugEnabled(true)
        .setOAuthConsumerKey("uryPvY5VNBPHHE6IeLPQvdQKU")
		.setOAuthConsumerSecret("B4nrhCnSHnEnywj61TXReqJnyG74ejyKa8V8jSVc0yv83wDloJ")
		.setOAuthAccessToken("222000067-qkuTMidVatLjCjD1aqEitZgXcOycxPJXRza7oM6n")
		.setOAuthAccessTokenSecret("DOgjuaV9fuG6N2H2LKeyMUeFEBRpzceoDHcv3XawL4HIX")
        .setJSONStoreEnabled(true);
        
        Twitter twitter = new TwitterFactory(cb.build()).getInstance();
        
        try {
        	//get labels

            File file = new File("/home/hduser/labels.txt");
            FileReader reader = new FileReader(file);
            BufferedReader bufferedReader = new BufferedReader(reader);
            String keywords = null;
            ArrayList list = new ArrayList();
            int index = 0;
            
            while((keywords = bufferedReader.readLine()) != null){
            	list.add(keywords);
            }
            
            for(int i =0; i<1001; i++){

            	//random generate keyword
            	int listindex = (int)(Math.random() * 1001);
            	String listkeywords = (String)list.get(listindex);
            	
	            String queryStr = listkeywords;
	            Query query = new Query(queryStr);
	            QueryResult result;
	            Boolean endflag = false;
	            int counter = 0;
	            
	            SimpleDateFormat sdf=new SimpleDateFormat("YYYY-MM-DD HH:MM:SS");  
	            do {
	                result = twitter.search(query);
	                List<Status> tweets = result.getTweets();
	                for (Status tweet : tweets) {

	                	Calendar starttime = Calendar.getInstance();
	                	Calendar endtime = Calendar.getInstance();
	                	starttime.add(Calendar.MINUTE, -2);
	                	Date date_start = starttime.getTime();
	                	Date date_end = endtime.getTime();
	                	String date_start_str=sdf.format(date_start);
	                	String date_end_str=sdf.format(date_end);
	                	

	                        String send = null;
	                        String language = "NA";
	                        
	                        if(tweet.getUser().getLang()!= null && tweet.getUser().getLang().length()!=0)
	                        	language = tweet.getUser().getLang();
	                        
	                        System.out.println("@" + tweet.getUser().getScreenName() + "||" + queryStr + "||" + language );
	                        
	                        send = "@" + tweet.getUser().getScreenName() + "||" + queryStr + "||" + language ;

	                        producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(index++), send));       
	                        counter++;
	                        if(counter > 20)
	                        {
	                        	endflag = true;
	                        }

	                }
	                Thread.sleep(5000);
	            } 
	            while ((query = result.nextQuery()) != null && endflag==false);
	            Thread.sleep(5000);
            }
            System.exit(0);
        } catch (Exception te) {
            te.printStackTrace();
            System.out.println("Failed to search tweets: " + te.getMessage());
            System.exit(-1);
        }
        System.out.println("Message sent successfully");
        producer.close();
    }
}
